package com.qupeng.demo.kafka.kafkaspringbootproducer.model;

import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import lombok.Data;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.support.SendResult;

import java.io.Serializable;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class ProduceSuccessResponse<K, V> implements Serializable {

    private ProducerRecordData<K, V> producerRecordData;

    private RecordMetaData recordMetaData;

    public ProduceSuccessResponse() {

    }

    public ProduceSuccessResponse(SendResult<K, V> sendResult) {
        producerRecordData = new ProducerRecordData<>(sendResult.getProducerRecord());
        recordMetaData = new RecordMetaData(sendResult.getRecordMetadata());
    }

    @Data
    private static class ProducerRecordData<K, V> implements Serializable {
        public ProducerRecordData(ProducerRecord<K, V> producerRecord) {
            this.topic = producerRecord.topic();
            this.partition = Optional.ofNullable(producerRecord.partition()).map(String::valueOf).orElse("null");
            Optional.ofNullable(producerRecord.headers()).ifPresentOrElse(headers -> {
                headers.forEach(header -> this.headers.put(header.key(), new String(header.value())));
            }, () -> {});
            this.key = producerRecord.key();
            this.value = producerRecord.value().toString();
            this.timestamp = Optional.ofNullable(producerRecord.timestamp()).map(Date::new).map(Date::toString).orElse("null");
        }

        String topic;
        String partition;
        Map<String, String> headers = new HashMap<>();
        K key;
        String value;
        String timestamp;
    }

    @Data
    private static class RecordMetaData implements Serializable {
        RecordMetaData(RecordMetadata recordMetadata) {
            this.offset = recordMetadata.offset();
            this.timestamp = new Date(recordMetadata.timestamp());
            this.serializedKeySize = recordMetadata.serializedKeySize();
            this.serializedValueSize = recordMetadata.serializedValueSize();
            this.partition = recordMetadata.partition();
        }
        long offset;
        @JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
        Date timestamp;
        int serializedKeySize;
        int serializedValueSize;
        int partition;
    }
}
